package cn.mesmile.flink.sink;

import cn.mesmile.flink.jdkstream.VideoOrder;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.Date;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author zb
 * @date 2021/8/21 20:15
 * @Description
 */
public class MysqlSink extends RichSinkFunction<VideoOrder> {

    private Connection conn = null;
    private PreparedStatement ps = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("----------mysql sink open---------");
        //1.加载驱动
        Class<?> aClass = Class.forName("com.mysql.cj.jdbc.Driver");
        // 2.获取连接
        conn = DriverManager.getConnection("jdbc:mysql://81.69.43.78:3306/flink?useUnicode=true&characterEncoding=utf8&allowMultiQueries=true&serverTimezone=Asia/Shanghai", "root", "task,.123");
        String sql = "INSERT INTO `video_order` (`user_id`, `money`, `title`, `trade_no`, `create_time`) VALUES(?,?,?,?,?);";
        // 3.预编译
        ps = conn.prepareStatement(sql);
    }

    @Override
    public void close() throws Exception {
        System.out.println("----------mysql sink close---------");
        // 释放连接
        if (conn != null) {
            conn.close();
        }
        if (ps != null) {
            ps.close();
        }
    }

    @Override
    public void invoke(VideoOrder videoOrder, Context context) throws Exception {
        //给ps中的?设置具体值
        ps.setInt(1,videoOrder.getUserId());
        ps.setInt(2,videoOrder.getMoney());
        ps.setString(3,videoOrder.getTitle());
        ps.setString(4,videoOrder.getTradeNo());
        ps.setDate(5,new Date(videoOrder.getCreateTime().getTime()));
        // 4.执行
        ps.executeUpdate();
    }
}
